package service

import (
	"context"
	"crypto/x509"
	"strings"
	"time"

	"hundsun.com/hsl/hschain/common/util"

	"github.com/golang/protobuf/ptypes/timestamp"
	websocket "github.com/libp2p/go-ws-transport"
	"google.golang.org/grpc"
	"google.golang.org/grpc/peer"
	"hundsun.com/hsl/hschain/common/log/logging"
	grpc2 "hundsun.com/hsl/hschain/common/util/grpc"
	"hundsun.com/hsl/hschain/protos"
	protosledger "hundsun.com/hsl/hschain/protos/ledger"
	protosrpc "hundsun.com/hsl/hschain/protos/rpc"
	"hundsun.com/hsl/hschain/rpc/common/constant"
	"hundsun.com/hsl/hschain/rpc/common/rpcerror"
	"hundsun.com/hsl/hschain/rpc/common/utils"
)

type server struct {
	module *RPCService
	protosrpc.RpcServer
	method     *RPCMethodImpl
	grpcServer *grpc.Server
}

//JSONRPCServer JSONRPCServer
type JSONRPCServer struct {
	*server
}

//JSONRPCArgs JSONRPCArgs
type JSONRPCArgs struct {
	FuncName string
	Param    []byte
}

//JSONRPCResp JSONRPCResp
type JSONRPCResp struct {
	Status  int32
	Message string
	Payload []byte
}

//ProcessRPCRequest ProcessRPCRequest
func (j JSONRPCServer) ProcessRPCRequest(args JSONRPCArgs, resp *JSONRPCResp) error {
	result, _ := j.server.ProcessRPCRequest(peer.NewContext(j.module.ctx, &peer.Peer{
		// todo jsonrpc获取真正的ip
		Addr:     websocket.NewAddr("0.0.0.0:57001"),
		AuthInfo: nil,
	}), &protosrpc.Request{
		Method: args.FuncName,
		Params: args.Param,
	})
	*resp = JSONRPCResp{
		Status:  result.Status,
		Message: result.Message,
		Payload: result.Payload,
	}
	return nil
}

//newRPCServerInstance module初始化时执行一次
func newRPCServerInstance(module *RPCService) *server {
	instance := NewRPCMethodInstance(module)
	serverInstance := &server{
		method:     instance,
		module:     module,
		grpcServer: grpc.NewServer(grpc2.GetDefaultServerOptions()...),
	}
	return serverInstance
}

//个性化签名校验，验证证书ip是否匹配
func (s *server) privateSignatureVerify(ctx context.Context, certificate *x509.Certificate) (bool, error) {
	ip, err := utils.GetClientIP(ctx)

	if err != nil {
		return false, err
	}

	if certificate.Subject.CommonName == strings.Split(ip, ":")[0] {
		return true, nil
	}

	return false, rpcerror.Errorf("ip :%s has no authority", ip)
}

func (s *server) rpcPreCheck(ctx context.Context) (*protosrpc.RpcResponse, error) {
	ip, err := utils.GetClientIP(ctx)
	if err != nil {
		return &protosrpc.RpcResponse{Status: rpcerror.CommonFail.GetStatus(), Message: err.Error()}, err
	}
	//流量控制
	if !s.FlowControl(ip) {
		return &protosrpc.RpcResponse{Status: rpcerror.FlowControlFailed.GetStatus(), Message: rpcerror.FlowControlFailed.GetMessage()}, nil
	}
	pass, err := s.AuthVerify(ctx, s.module.rpcConfig.GetParentCert(), s.privateSignatureVerify)
	if err != nil {
		return &protosrpc.RpcResponse{Status: rpcerror.AuthFailed.GetStatus(), Message: err.Error()}, err
	}
	if !pass {
		return &protosrpc.RpcResponse{Status: rpcerror.AuthFailed.GetStatus(), Message: rpcerror.AuthFailed.GetMessage()}, nil
	}
	return nil, nil
}

func (s *server) rpcCall(ctx context.Context, req *protosrpc.Request) (*protosrpc.RpcResponse, error) {
	//设置请求时间
	req.Timestamp = &timestamp.Timestamp{}
	req.Timestamp.Seconds = time.Now().Unix()

	if req == nil {
		return nil, rpcerror.Errorf("nil parameter for ProcessRpcRequest")
	}

	handler := GetEventHandler(req.GetMethod())
	if handler == nil {
		return &protosrpc.RpcResponse{
			Payload: nil, Timestamp: &timestamp.Timestamp{}, Status: rpcerror.MethodNotFound.GetStatus(), Message: rpcerror.MethodNotFound.GetMessage()}, nil
	}
	res, code, err := handler(ctx, req)
	if err != nil {
		return &protosrpc.RpcResponse{Payload: nil, Timestamp: &timestamp.Timestamp{}, Status: code, Message: err.Error()}, nil
	}
	buf, err := util.Struct2Bytes(res)
	if err != nil {
		return &protosrpc.RpcResponse{Payload: nil, Timestamp: &timestamp.Timestamp{}, Status: code, Message: err.Error()}, nil
	}
	return &protosrpc.RpcResponse{Payload: buf, Timestamp: &timestamp.Timestamp{}, Status: code, Message: ""}, nil
}

//ProcessRPCRequest ProcessRpcRequest
func (s *server) ProcessRPCRequest(ctx context.Context, req *protosrpc.Request) (*protosrpc.RpcResponse, error) {
	s.module.Log.Tracef("grpc call method:%s, param:%v", req.GetMethod(), req)
	resp, err := s.rpcPreCheck(ctx)
	if err != nil {
		return resp, err
	}
	return s.rpcCall(ctx, req)
}

//ProcessTxSend ProcessTxSend
func (s *server) ProcessTxSend(stream protosrpc.Rpc_ProcessTxSendServer) error {
	//验签
	pass, err := s.AuthVerify(stream.Context(), s.module.rpcConfig.GetParentCert(), s.privateSignatureVerify)
	if err != nil || !pass {
		return err
	}
	ip, _ := utils.GetClientIP(stream.Context())

	//存在则更新stream， 不存在则新建
	var sleepTime = 0
	ticker := time.NewTicker(time.Second * constant.StreamExpireCheckPeriod)
	ctx, cancelFun := context.WithCancel(stream.Context())
	go func() {
		defer func() {
			s.module.Log.Infof("receive close msg from tx stream, endpoint:%s", ip)
		}()
		for {
			select {
			case <-ctx.Done():
				return
			default:
				var in = protosrpc.Request{}
				recvErr := stream.RecvMsg(&in)
				//流量控制
				if !s.FlowControl(ip) {
					_ = stream.Send(&protosrpc.RpcResponse{Status: rpcerror.FlowControlFailed.GetStatus(), Message: rpcerror.FlowControlFailed.GetMessage()})
				}
				sleepTime = 0
				if recvErr != nil {
					s.module.Log.Errorf("receive tx stream failed, reason:%s", recvErr.Error())
					return
				}
				//如果是心跳，则continue
				if in.GetParams() == nil {
					s.module.Log.Debugf("received tx heart beat, endpoint:%s", ip)
					continue
				}
				var methodImpl = s.method
				var obj = &protosledger.Transaction{}
				objI, err := util.Bytes2Struct(obj, in.GetParams())
				if err != nil {
					s.module.Log.Errorf("deserialize transaction failed, reason:%s", err.Error())
					_ = stream.Send(&protosrpc.RpcResponse{Status: rpcerror.SerializeFail.GetStatus(), Message: err.Error()})
					continue
				}
				hash, code, err := methodImpl.SendSignedTransaction(objI.(*protosledger.Transaction))
				var msg string
				if err != nil {
					msg = err.Error()
				}
				_ = stream.Send(&protosrpc.RpcResponse{Payload: hash.(protos.Hash), Timestamp: &timestamp.Timestamp{}, Status: code, Message: msg})
			}
		}
	}()

	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			sleepTime = sleepTime + constant.StreamExpireCheckPeriod
			if sleepTime > constant.StreamExpireTime {
				ip, _ := utils.GetClientIP(stream.Context())
				s.module.Log.Infof("tx stream expired, endpoint:%s", ip)
				ticker.Stop()
				cancelFun()
				return nil
			}
		}
	}
}

//ProcessSubRequest ProcessSubRequest
func (s *server) ProcessSubRequest(stream protosrpc.Rpc_ProcessSubRequestServer) error {

	//验签
	pass, err := s.AuthVerify(stream.Context(), s.module.rpcConfig.GetParentCert(), s.privateSignatureVerify)
	if err != nil || !pass {
		return err
	}

	//var heartBeat = make(chan int32)
	newStream := &StreamEnhance{
		Rpc_ProcessSubRequestServer: stream,
		Disable:                     constant.Normal,
		SleepTime:                   0,
	}
	ctx, cancelFun := context.WithCancel(stream.Context())
	ip, _ := utils.GetClientIP(stream.Context())
	processLog := s.module.Log.WithField(logging.LogFieldRPCEndpoint, ip)
	ticker := time.NewTicker(constant.StreamExpireCheckPeriod * time.Second)
	//存在则更新stream， 不存在则新建
	_, err = s.module.endpointInstance.NewEndpoint(newStream)
	if err != nil {
		return err
	}

	closeFun := func() {
		ticker.Stop()
		newStream.Disable = constant.Abnormal
		cancelFun()
	}

	go func() {
		defer func() {
			closeFun()
		}()
		for {
			select {
			case <-ctx.Done():
				processLog.Infof("stream closed by client")
				return
			default:
				var in = protosrpc.Request{}
				recvErr := stream.RecvMsg(&in)
				//流量控制
				if !s.FlowControl(ip) {
					_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
						{Status: rpcerror.FlowControlFailed.GetStatus(), Message: rpcerror.FlowControlFailed.GetMessage()}}})
				}
				newStream.SleepTime = 0
				if recvErr != nil {
					return
				}
				//如果是心跳，则continue
				if in.GetParams() == nil {
					s.module.Log.Debugf("received subscribe heart beat, endpoint:%s", ip)
					continue
				}
				//设置请求时间
				in.Timestamp = &timestamp.Timestamp{}
				in.Timestamp.Seconds = time.Now().Unix()
				//区分不同订阅
				method, _, err := s.method.GetSubMethodAndParam(in.GetMethod())

				if err != nil {
					_ = stream.Send(&protosrpc.SubscriptionResult{Result: []*protosrpc.Result{
						{Status: rpcerror.MethodNotFound.GetStatus(), Message: rpcerror.MethodNotFound.GetMessage()}}})
				} else {
					//并发处理订阅请求
					utils.Invoke(method, []interface{}{&in, stream})
				}
			}
		}
	}()
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
			newStream.SleepTime = newStream.SleepTime + constant.StreamExpireCheckPeriod
			if newStream.SleepTime > constant.StreamExpireTime {
				processLog.Infof("stream time out and close")
				closeFun()
				return nil
			}
		}
	}
}

//FlowControl 流量控制
func (s *server) FlowControl(ip string) bool {
	return s.module.flowControlInstance.Check(ip)
}

//AuthVerify rpc模块grpc验证
func (s *server) AuthVerify(ctx context.Context, caBuf []byte, privateVerify func(ctx context.Context, certificate *x509.Certificate) (bool, error)) (bool, error) {
	if !s.module.rpcConfig.NeedAuthVerify() {
		return true, nil
	}
	return grpc2.AuthVerify(ctx, caBuf, privateVerify)
}
